package com.wpml.netty.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

public class SelectorThreadGroup {

    SelectorThread[] sts;
    ServerSocketChannel server = null;
    AtomicInteger xid = new AtomicInteger(0);

    SelectorThreadGroup worker = this;

    public SelectorThreadGroup(int num) {
        sts = new SelectorThread[num];

        for (int i = 0; i < num; i++) {
            sts[i] = new SelectorThread(this);

            new Thread(sts[i]).start();
        }
    }

    public void bind(int port) {
        try {
            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));


            nextSelectorV2(server);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void nextSelector(Channel server) {
        SelectorThread st = next();
        st.queue.add(server);
        st.selector.wakeup();
    }

    private SelectorThread next() {
        int index = xid.incrementAndGet() % sts.length;
        return sts[index];
    }

    public void nextSelectorV2(Channel c) {
        // boss 与 worker 单独分配线程组
        if (c instanceof ServerSocketChannel) {
            // listen 选择了 boss组中的一个线程后，要更新这个线程的work组
            SelectorThread boss = this.next();
            boss.queue.add(c);
            // 设置实际工作线程组
            boss.setWorker(worker);
            boss.selector.wakeup();
        } else {
            SelectorThread st = nextV2();
            st.queue.add(c);
            st.selector.wakeup();
        }
    }

    private SelectorThread nextV2() {
        int index = xid.incrementAndGet() % worker.sts.length;
        return worker.sts[index];
    }

    // 工作线程组
    public void setWorker(SelectorThreadGroup worker) {
        this.worker = worker;
    }
}
